package com.bff.gaia.mix.api.xjoin2.operators;

import com.bff.gaia.api.java.typeutils.RowTypeInfo;
import com.bff.gaia.mix.api.datastream.AsyncDataMixStream;
import com.bff.gaia.mix.api.datastream.DataMixStream;
import com.bff.gaia.mix.api.datastream.SingleOutputMixStreamOperator;
import com.bff.gaia.mix.api.xjoin2.client.mysql.async.MysqlRdbRichAsyncXjoinFunction;
import com.bff.gaia.mix.api.xjoin2.core.side.SideTableInfo;
import com.bff.gaia.mix.api.xjoin2.core.table.FieldInfo;
import com.bff.gaia.mix.api.xjoin2.core.table.JoinInfo;
import com.bff.gaia.mix.api.xjoin2.core.table.StreamTableInfo;
import com.bff.gaia.mix.api.xjoin2.functions.RichAsyncXjoinFunction;
import com.bff.gaia.types.Row;

import java.util.List;
import java.util.concurrent.TimeUnit;

public class AsyncXjoinOperatorGateWay {
	private static RichAsyncXjoinFunction loadRichAsyncXjoinFunction(
		StreamTableInfo streamTableInfo,
		SideTableInfo sideTableInfo,
		JoinInfo joinInfo,
		List<FieldInfo> outFieldInfoList) {
		String sideType = sideTableInfo.getType();
		switch (sideType.toLowerCase()) {
			case "mysql": {
				return new MysqlRdbRichAsyncXjoinFunction(streamTableInfo, sideTableInfo, joinInfo, outFieldInfoList);
			}
			default: {
				return null;
			}
		}
	}


	public static SingleOutputMixStreamOperator<Row> getAsyncXjoinStream(DataMixStream<Row> stream,
	                                                                   StreamTableInfo streamTableInfo,
	                                                                   SideTableInfo sideTableInfo,
	                                                                   JoinInfo joinInfo,
	                                                                   List<FieldInfo> outFieldInfoList,
	                                                                   RowTypeInfo outTypeInfo
	                                                                   ) {
RichAsyncXjoinFunction asyncXjoinFunction=loadRichAsyncXjoinFunction(streamTableInfo, sideTableInfo, joinInfo, outFieldInfoList);
		return AsyncDataMixStream
			.orderedWait(stream, asyncXjoinFunction, joinInfo.getCacheTimeout(), TimeUnit.MILLISECONDS, joinInfo.getAsyncCapacity())
			.setParallelism(joinInfo.getParallelism())
			.returns(outTypeInfo);
	}
}